package com.hopu.shop;

import com.alibaba.fastjson.JSONObject;

import com.hopu.bean.shop.ShopTop10;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class ShopTop10Ana {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<ShopTop10, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String sid = json.getString("shop_id");
            String sname = json.getString("shop_name");
            String type = json.getString("event_type");

            return new Tuple2<>(new ShopTop10(sid, sname, type, 0), 1);
        });

        JavaPairRDD<ShopTop10, Integer> viewRDD = pairRDD.filter(t -> "view".equals(t._1.getType()));
        groupBySid(viewRDD);
        JavaPairRDD<ShopTop10, Integer> cartRDD = pairRDD.filter(t -> "cart".equals(t._1.getType()));
        groupBySid(cartRDD);
        JavaPairRDD<ShopTop10, Integer> parchaseRDD = pairRDD.filter(t -> "parchase".equals(t._1.getType()));
        groupBySid(parchaseRDD);
    }

    public static void groupBySid(JavaPairRDD<ShopTop10, Integer> rdd) {
        JavaPairRDD<ShopTop10, Iterable<Integer>> groupBySid = rdd.groupByKey();

        JavaPairRDD<ShopTop10, Integer> group = groupBySid.mapToPair(t -> {
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            t._1.setCount(count);
            return new Tuple2<>(t._1, count);
        });

        List<ShopTop10> top10s = group.sortByKey().map(t -> t._1).take(10);
        top10s.forEach(t -> System.out.println(t));

        String[] columns = {"sid", "sname", "count"};

        for (ShopTop10 top10: top10s) {

            String[] values = {top10.getSid(), top10.getSname(), top10.getCount()+""};
            String family = top10.getType().equals("view")? "top10view": top10.getType().equals("cart")? "top10cart": "top10parchase";
            try {
                HBaseUtils.putOneRowToHbase("shopana", top10.getSid(), family, columns, values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


}
